package com.wzw.springboot.principle.use.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.CollectionUtils;

import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author Wangzhiwen
 */
@Slf4j
public class CanalClient {

    @SuppressWarnings("all")
    public void start() {
        // destination对应的是canal server的instance
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
        int batchSize = 5 * 1024;
        try {
            canalConnector.connect();
            canalConnector.subscribe(".*\\..*");
            while (true) {
                Message message = canalConnector.getWithoutAck(batchSize);
                // 本地获取的消息ID
                long batchId = message.getId();
                // 数据量
                int size = message.getEntries().size();
                // log.info("MessageId[{}], Size[{}]", batchId, size);
                if (batchId != -1 && size > 0) {
                    printSummary(message, batchId, size);
                    printEntry(message.getEntries());
                    canalConnector.ack(batchId);
                }
            }
        } catch (Exception e) {
            canalConnector.rollback();
            log.error("", e);
        } finally {
            canalConnector.disconnect();
        }
    }

    String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";

    protected void printSummary(Message message, long batchId, int size) {
        long memsize = 0;
        for (CanalEntry.Entry entry : message.getEntries()) {
            memsize += entry.getHeader().getEventLength();
        }

        String startPosition = null;
        String endPosition = null;
        if (!CollectionUtils.isEmpty(message.getEntries())) {
            startPosition = buildPositionForDump(message.getEntries().get(0));
            endPosition = buildPositionForDump(message.getEntries().get(message.getEntries().size() - 1));
        }

        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        log.info("context_format : {} {} {} {} {} {}", batchId, size, memsize, format.format(new Date()), startPosition,
                endPosition);
    }

    protected String buildPositionForDump(CanalEntry.Entry entry) {
        long time = entry.getHeader().getExecuteTime();
        Date date = new Date(time);
        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        String position = entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
                + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
        if (StringUtils.isNotEmpty(entry.getHeader().getGtid())) {
            position += " gtid(" + entry.getHeader().getGtid() + ")";
        }
        return position;
    }

    protected void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = System.currentTimeMillis() - executeTime;
            Date date = new Date(entry.getHeader().getExecuteTime());
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN) {
                    CanalEntry.TransactionBegin begin;
                    try {
                        begin = CanalEntry.TransactionBegin.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务头信息，执行的线程id，事务耗时
                    log.info("transaction_format : {} {} {} {} {} {}",
                            entry.getHeader().getLogfileName(),
                            entry.getHeader().getLogfileOffset(),
                            entry.getHeader().getExecuteTime(), simpleDateFormat.format(date),
                            entry.getHeader().getGtid(), delayTime);
                    log.info(" BEGIN ----> Thread id: {}", begin.getThreadId());
                    printXAInfo(begin.getPropsList());
                } else {
                    CanalEntry.TransactionEnd end;
                    try {
                        end = CanalEntry.TransactionEnd.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务提交信息，事务id
                    log.info(" END ----> transaction id: {}", end.getTransactionId());
                    printXAInfo(end.getPropsList());
                    log.info("transaction_format : {} {} {} {} {} {}",
                            entry.getHeader().getLogfileName(),
                            entry.getHeader().getLogfileOffset(),
                            entry.getHeader().getExecuteTime(), simpleDateFormat.format(date),
                            entry.getHeader().getGtid(), delayTime);
                }
                continue;
            }

            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChange;
                try {
                    rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                }

                CanalEntry.EventType eventType = rowChange.getEventType();

                log.info("row_format : {} {} {} {} {} {} {} {} {}",
                        entry.getHeader().getLogfileName(),
                        entry.getHeader().getLogfileOffset(), entry.getHeader().getSchemaName(),
                        entry.getHeader().getTableName(), eventType,
                        entry.getHeader().getExecuteTime(), simpleDateFormat.format(date),
                        entry.getHeader().getGtid(), delayTime);

                if (eventType == CanalEntry.EventType.QUERY || rowChange.getIsDdl()) {
                    log.info(" sql ----> " + rowChange.getSql() + System.lineSeparator());
                    continue;
                }

                printXAInfo(rowChange.getPropsList());
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    if (eventType == CanalEntry.EventType.DELETE) {
                        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                        String collect = beforeColumnsList.stream().map(it -> it.getName() + "=" + it.getValue()).collect(Collectors.joining(","));
                        log.info(" sql ----> DELETE, {}", collect);
                        // curdContext.executeDelete(entry.getHeader().getTableName(), rowData);
                    } else if (eventType == CanalEntry.EventType.INSERT) {
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        String collect = afterColumnsList.stream().map(it -> it.getName() + "=" + it.getValue()).collect(Collectors.joining(","));
                        log.info(" sql ----> INSERT, {}", collect);
                        // 表名称
                        String tableName = entry.getHeader().getTableName();
                        // 字段信息
                        List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
                        for (CanalEntry.Column column : columns) {
                            // 字段名称
                            String name = column.getName();
                            // 字段值
                            String value = column.getValue();
                        }
                        // curdContext.executeInsert(entry.getHeader().getTableName(), rowData);
                    } else {
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        String collect = afterColumnsList.stream().map(it -> it.getName() + "=" + it.getValue()).collect(Collectors.joining(","));
                        log.info(" sql ----> UPDATE, {}", collect);
                        // curdContext.executeUpdate(entry.getHeader().getTableName(), rowData);
                    }
                }
            }
        }
    }

    protected void printXAInfo(List<CanalEntry.Pair> pairs) {
        if (pairs == null) {
            return;
        }

        String xaType = null;
        String xaXid = null;
        for (CanalEntry.Pair pair : pairs) {
            String key = pair.getKey();
            if (StringUtils.endsWithIgnoreCase(key, "XA_TYPE")) {
                xaType = pair.getValue();
            } else if (StringUtils.endsWithIgnoreCase(key, "XA_XID")) {
                xaXid = pair.getValue();
            }
        }

        if (xaType != null && xaXid != null) {
            log.info(" ------> " + xaType + " " + xaXid);
        }
    }

}
